package com.tyc.threalpool;

import java.util.ArrayList;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * -Xmx10m -Xms10m
 * 原文： https://mp.weixin.qq.com/s/id_3pMbUi2F-6PqQ9Qosyw
 * 产生原因：自定义拒绝策略时没有抛出异常，当提交futureTask任务且触发拒绝策略时，
 * submit方法返回的futureTask对象状态为normal状态，但任务实际上并没有提交成功。
 * 导致调用futureTask.get时会被LockSupport.park(this)一直阻塞住
 * @author tyc
 * @version 1.0
 * @date 2023-01-10 16:12:13
 */
public class Demo04 {
    public static void main(String[] args) throws InterruptedException {
        ThreadPoolExecutor productThreadPoolExecutor = new ThreadPoolExecutor(1,
                1,
                1,
                TimeUnit.SECONDS,
                new LinkedBlockingQueue<>(1),
                new MyFactory("product"),
                new MyRejectPolicy());

        while (true){
            TimeUnit.SECONDS.sleep(1);
            new Thread(()->{
                ArrayList<Future<Integer>> futureList = new ArrayList<>();
                //从数据库获取产品信息
                int productNum = 5;
                for (int i = 0; i < productNum; i++) {
                    try {
                        int finalI = i;
                        Future<Integer> future = productThreadPoolExecutor.submit(() -> {
                            System.out.println("Thread.currentThread().getName() = " + Thread.currentThread().getName());
                            return finalI * 10;
                        });
                        futureList.add(future);
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
                for (Future<Integer> integerFuture : futureList) {
                    try {
                        Integer integer = integerFuture.get();
                        System.out.println(integer);
                        System.out.println("future.get() = " + integer);
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            }).start();
        }
    }

    /**
     * 它和默认的线程工厂之间唯一的区别就是我加了一个 threadFactoryName 字段，方便给线程池里面的线程取一个合适的名字。
     * 原生：pool-1-thread-1
     * 自定义：product-pool-1-thread-1
     */
    static class MyFactory implements ThreadFactory{
        private static final AtomicInteger poolNumber = new AtomicInteger(1);

        private final ThreadGroup group;

        private final AtomicInteger threadNum = new AtomicInteger(1);

        private final String namePrefix;

        private final String threadFactoryName;

        public String getThreadFactoryName(){
            return threadFactoryName;
        }


        MyFactory(String threadStartName){
            SecurityManager s = System.getSecurityManager();
            group = s != null ? s.getThreadGroup() : Thread.currentThread().getThreadGroup();
            namePrefix = threadStartName + "-pool-" + poolNumber.getAndIncrement()  + "-thread";
            threadFactoryName = threadStartName;
        }

        @Override
        public Thread newThread(Runnable r) {
            Thread t = new Thread(group,r,namePrefix+threadNum.getAndIncrement(),0);
            if(t.isDaemon()){
                t.setDaemon(false);
            }
            if(t.getPriority() != Thread.NORM_PRIORITY){
                t.setPriority(Thread.NORM_PRIORITY);
            }
            return t;
        }
    }

    static class MyRejectPolicy implements RejectedExecutionHandler{

        @Override
        public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
            if(executor.getThreadFactory() instanceof MyFactory){
                MyFactory factory = (MyFactory)executor.getThreadFactory();
                if("product".equals(factory.getThreadFactoryName())){
                    System.out.println("product线程池有任务被拒绝");
                }
            }
            // 如果未抛出异常会造成内存泄漏
            throw new RejectedExecutionException("task"+r.toString()+"rejected from +"+executor.toString());
        }
    }
}
